package vip.zhenzicheng.rocketmq.config;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.StringMessageConverter;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 全局事务消息监听器
 *
 * @author zhenzicheng
 * @date 2022-05-29 13:13
 */
// @RocketMQTransactionListener(txProducerGroup = "springBootGroup2")
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
public class MyTransactionImpl implements RocketMQLocalTransactionListener {

  private final ConcurrentHashMap<Object, Message> localTrans = new ConcurrentHashMap<>();

  @Override
  public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
    Object transId = msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
    String destination = arg.toString();
    localTrans.put(transId, msg);
    //这个msg的实现类是GenericMessage，里面实现了toString方法
    //在Header中自定义的RocketMQHeaders.TAGS属性，到这里就没了。但是RocketMQHeaders.TRANSACTION_ID这个属性就还在。
    //而message的Header里面会默认保存RocketMQHeaders里的属性，但是都会加上一个RocketMQHeaders.PREFIX前缀
    System.out.println("executeLocalTransaction msg = " + msg);
    //转成RocketMQ的Message对象
    org.apache.rocketmq.common.message.Message message = RocketMQUtil.convertToRocketMessage(new StringMessageConverter(), "UTF-8", destination, msg);
    String tags = message.getTags();
    if (StringUtils.contains(tags, "TagA")) {
      return RocketMQLocalTransactionState.COMMIT;
    } else if (StringUtils.contains(tags, "TagB")) {
      return RocketMQLocalTransactionState.ROLLBACK;
    } else {
      return RocketMQLocalTransactionState.UNKNOWN;
    }
  }

  // 延迟检查的时间间隔要有点奇怪。
  @Override
  public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
    String transId = msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID).toString();
    Message originalMessage = localTrans.get(transId);
    //这里能够获取到自定义的transaction_id属性
    System.out.println("checkLocalTransaction msg = " + originalMessage);
    //获取标签时，自定义的RocketMQHeaders.TAGS拿不到，但是框架会封装成一个带RocketMQHeaders.PREFIX的属性
//        String tags = msg.getHeaders().get(RocketMQHeaders.TAGS).toString();
    String tags = msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TAGS).toString();
    if (StringUtils.contains(tags, "TagC")) {
      return RocketMQLocalTransactionState.COMMIT;
    } else if (StringUtils.contains(tags, "TagD")) {
      return RocketMQLocalTransactionState.ROLLBACK;
    } else {
      return RocketMQLocalTransactionState.UNKNOWN;
    }
  }
}
